package io.kiki.stack.netty.invoker;

import io.kiki.stack.netty.command.AbstractResponseCommand;
import io.kiki.stack.netty.command.ResponseCommand;
import io.kiki.stack.netty.command.ResponseType;
import io.kiki.stack.netty.exception.*;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;


@Slf4j
@Setter
@Getter
public class CallbackTask implements Runnable {

    Future future;
    String remoteAddress;

    public CallbackTask(String remoteAddress, Future future) {
        this.remoteAddress = remoteAddress;
        this.future = future;
    }

    @Override
    public void run() {
        Callback callback = future.getInvokeCallback();
        // a lot of try-catches to protect thread pool
        ResponseCommand responseCommand = null;

        try {
            responseCommand = (ResponseCommand) future.waitResponse(0);
        } catch (InterruptedException e) {
            String msg = "Exception caught when getting response from InvokeFuture. The address is " + this.remoteAddress;
            log.error(msg, e);
        }
        if (responseCommand == null || responseCommand.getResponseStatus() != ResponseType.success) {
            try {
                Exception e;
                if (responseCommand == null) {
                    e = new InvokeException("Exception caught in invocation. The address is " + this.remoteAddress + " responseStatus:" + ResponseType.unknown, future.getCause());
                } else {
                    responseCommand.setRequestContext(future.getInvokeContext());
                    switch (responseCommand.getResponseStatus()) {
                    case timeout:
                        e = new InvokeTimeoutException("Invoke timeout when invoke with callback.The address is " + this.remoteAddress);
                        break;
                    case channel_closed:
                        e = new ChannelClosedException("channel closed when invoke with callback.The address is " + this.remoteAddress);
                        break;
                    case server_threadpool_busy:
                        e = new InvokeServerBusyException("Server thread pool busy when invoke with callback.The address is " + this.remoteAddress);
                        break;
                    case server_exception:
                        String msg = "Server exception when invoke with callback.Please check the server log! The address is " + this.remoteAddress;
                        AbstractResponseCommand resp = (AbstractResponseCommand) responseCommand;
                        resp.deserialize();
                        Object ex = resp.getResponseObject();
                        if (ex instanceof Throwable) {
                            e = new InvokeServerException(msg, (Throwable) ex);
                        } else {
                            e = new InvokeServerException(msg);
                        }
                        break;
                    default:
                        e = new InvokeException("Exception caught in invocation. The address is " + this.remoteAddress + " responseStatus:" + responseCommand.getResponseStatus(), future.getCause());

                    }
                }
                callback.onFailed(e);
            } catch (Throwable e) {
                log.error("Exception occurred in user defined InvokeCallback#onException() logic, The address is {}", this.remoteAddress, e);
            }
        } else {
            ClassLoader oldClassLoader = null;
            try {
                if (future.getAppClassLoader() != null) {
                    oldClassLoader = Thread.currentThread().getContextClassLoader();
                    Thread.currentThread().setContextClassLoader(future.getAppClassLoader());
                }
                responseCommand.setRequestContext(future.getInvokeContext());
                AbstractResponseCommand rpcResponse = (AbstractResponseCommand) responseCommand;
                responseCommand.deserialize();
                try {
                    callback.onSuccess(rpcResponse.getResponseObject());
                } catch (Throwable e) {
                    log.error("Exception occurred in user defined InvokeCallback#onResponse() logic.", e);
                }
            } catch (CodecException e) {
                log.error("CodecException caught on when deserialize response in RpcInvokeCallbackListener. The address is {}.", this.remoteAddress, e);
            } catch (Throwable e) {
                log.error("Exception caught in RpcInvokeCallbackListener. The address is {}", this.remoteAddress, e);
            } finally {
                if (oldClassLoader != null) {
                    Thread.currentThread().setContextClassLoader(oldClassLoader);
                }
            }
        } // enf of else
    } // end of run
}
